package com.ruoyi.flink.wc.util;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Random;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import com.ruoyi.flink.util.db.ClickHouseLink;

public class ParticiplesWrite extends RichSinkFunction<Tuple2<String, Long>>{

	private static final long serialVersionUID = 1L;

	Connection connection = null;

	    String sql;

	    public ParticiplesWrite(String sql) {
	        this.sql = sql;
	    }

	    @Override
	    public void open(Configuration parameters) throws Exception {
	        super.open(parameters);
	        connection = ClickHouseLink.getConn();
	    }

	    @Override
	    public void close() throws Exception {
	        super.close();
	        if (connection != null) {
	            connection.close();
	        }
	    }

	    @Override
	    public void invoke(Tuple2<String, Long> tup, Context context) throws Exception {
	        PreparedStatement preparedStatement = connection.prepareStatement(sql);
	        preparedStatement.setLong(1, System.currentTimeMillis()+ new Random().nextInt(10));
	        preparedStatement.setString(2, tup.f0);
	        preparedStatement.setLong(3, tup.f1);
	        preparedStatement.setLong(4, System.currentTimeMillis());
	        preparedStatement.execute();
	        //connection.commit();
	        System.out.println("完毕" );
	    }
}
